package com.rrd.dw.mr.gzip.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.third.baihang.BaihangC1Data;
import com.ucredit.babel.common.model.third.baihang.BaihangC1Record;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;

/**
 * 百行征信
 *
 * @author rrz
 */

public class BaihangC1Parse extends ParseBase {
    private Text val = new Text();
    private Text key = new Text();

    /**
     * 百行征信
     *
     * @param model
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        BaihangC1Record obj = (BaihangC1Record) model;
        BaihangC1Data data = obj.getData();
        JSONObject info = data.getData() == null ? new JSONObject() : JSON.parseObject(data.getData());
        JSONObject D30 = info.getJSONObject("revolvingLoan").getJSONObject("D30");
        JSONObject summary = info.getJSONObject("revolvingLoan").getJSONObject("summary");
        JSONObject D360 = info.getJSONObject("revolvingLoan").getJSONObject("D360");
        JSONObject D180 = info.getJSONObject("revolvingLoan").getJSONObject("D180");
        JSONObject D90 = info.getJSONObject("revolvingLoan").getJSONObject("D90");
        JSONObject ND30 = info.getJSONObject("nonRevolvingLoan").getJSONObject("D30");
        JSONObject Nsummary = info.getJSONObject("nonRevolvingLoan").getJSONObject("summary");
        JSONObject ND360 = info.getJSONObject("nonRevolvingLoan").getJSONObject("D360");
        JSONObject ND180 = info.getJSONObject("nonRevolvingLoan").getJSONObject("D180");
        JSONObject ND90 = info.getJSONObject("nonRevolvingLoan").getJSONObject("D90");
        JSONObject ReportHeader = info.getJSONObject("reportHeader");
        JSONObject PersonalProfile = info.getJSONObject("personalProfile");

        StringBuilder sb = getBaseInfo(obj, successful);
        //循环授信-未结清总余额
        sb.append(summary.getString("remainingAmount")).append(CRT001);
        //循环授信-当前逾期金额
        sb.append(summary.getString("remainingOverdueAmount")).append(CRT001);
        //循环授信-累计账户数
        sb.append(summary.getString("accountCount")).append(CRT001);
        //循环授信-当前最严重逾期状态
        sb.append(summary.getString("remainingMaxOverdueStatus")).append(CRT001);
        //循环授信-当前授信总额
        sb.append(summary.getString("creditLimitSum")).append(CRT001);
        //循环授信-有效循环授信账户数
        sb.append(summary.getString("validAccountCount")).append(CRT001);
        //循环授信-当前逾期循环授信账户数
        sb.append(summary.getString("remainingOverdueAccountCount")).append(CRT001);
        //循环授信-历史最严重逾期状态
        sb.append(summary.getString("maxOverdueStatus")).append(CRT001);
        //循环授信-单个机构最高授信额度
        sb.append(summary.getString("maxCreditLimitPerTenant")).append(CRT001);
        //循环授信-累计逾期次数
        sb.append(summary.getString("overdueCount")).append(CRT001);
        //循环授信-最近一次发生代偿日期
        sb.append(summary.getString("revolvingLastCompensationDate")).append(CRT001);
        //循环授信-近30天新增循环授信账户数
        sb.append(D30.getString("accountCount")).append(CRT001);
        //循环授信-近30天新增循环授信额度
        sb.append(D30.getString("creditLimitSum")).append(CRT001);
        //循环授信-近30天发生逾期账户数
        sb.append(D30.getString("overdueAccountCount")).append(CRT001);
        //循环授信-近30天申请机构数量
        sb.append(D30.getString("applyTenantCount")).append(CRT001);
        //循环授信-近30天新增借款金额
        sb.append(D30.getString("lendingAmount")).append(CRT001);
        //循环授信-近30天发生代偿账户数
        sb.append(D30.getString("revolvingCompensationCount")).append(CRT001);
        //循环授信-近30天发生代偿次数
        sb.append(D30.getString("revolvingCompensationTimes")).append(CRT001);
        //循环授信-近30天发生代偿总金额
        sb.append(D30.getString("revolvingCompensationAmount")).append(CRT001);
        //循环授信-近90天新增循环授信账户数
        sb.append(D90.getString("accountCount")).append(CRT001);
        //循环授信-近90天新增循环授信额度
        sb.append(D90.getString("creditLimitSum")).append(CRT001);
        //循环授信-近90天发生逾期账户数
        sb.append(D90.getString("overdueAccountCount")).append(CRT001);
        //循环授信-近90天申请机构数量
        sb.append(D90.getString("applyTenantCount")).append(CRT001);
        //循环授信-近90天新增借款金额
        sb.append(D90.getString("lendingAmount")).append(CRT001);
        //循环授信-近90天发生代偿账户数
        sb.append(D90.getString("revolvingCompensationCount")).append(CRT001);
        //循环授信-近90天发生代偿次数
        sb.append(D90.getString("revolvingCompensationTimes")).append(CRT001);
        //循环授信-近90天发生代偿总金额
        sb.append(D90.getString("revolvingCompensationAmount")).append(CRT001);
        //循环授信-近180天新增循环授信账户数
        sb.append(D180.getString("accountCount")).append(CRT001);
        //循环授信-近180天新增循环授信额度
        sb.append(D180.getString("creditLimitSum")).append(CRT001);
        //循环授信-近180天发生逾期账户数
        sb.append(D180.getString("overdueAccountCount")).append(CRT001);
        //循环授信-近180天申请机构数量
        sb.append(D180.getString("applyTenantCount")).append(CRT001);
        //循环授信-近180天新增借款金额
        sb.append(D180.getString("lendingAmount")).append(CRT001);
        //循环授信-近180天发生代偿账户数
        sb.append(D180.getString("revolvingCompensationCount")).append(CRT001);
        //循环授信-近180天发生代偿次数
        sb.append(D180.getString("revolvingCompensationTimes")).append(CRT001);
        //循环授信-近180天发生代偿总金额
        sb.append(D180.getString("revolvingCompensationAmount")).append(CRT001);
        //循环授信-近360天新增循环授信账户数
        sb.append(D360.getString("accountCount")).append(CRT001);
        //循环授信-近360天新增循环授信额度
        sb.append(D360.getString("creditLimitSum")).append(CRT001);
        //循环授信-近360天发生逾期账户数
        sb.append(D360.getString("overdueAccountCount")).append(CRT001);
        //循环授信-近360天申请机构数量
        sb.append(D360.getString("applyTenantCount")).append(CRT001);
        //循环授信-近360天新增借款金额
        sb.append(D360.getString("lendingAmount")).append(CRT001);
        //循环授信-近360天发生代偿账户数
        sb.append(D360.getString("revolvingCompensationCount")).append(CRT001);
        //循环授信-近360天发生代偿次数
        sb.append(D360.getString("revolvingCompensationTimes")).append(CRT001);
        //循环授信-近360天发生代偿总金额
        sb.append(D360.getString("revolvingCompensationAmount")).append(CRT001);
        //非循环贷款-未结清总余额
        sb.append(Nsummary.getString("remainingAmount")).append(CRT001);
        //非循环贷款-当前逾期贷款笔数
        sb.append(Nsummary.getString("remainingOverdueLoanCount")).append(CRT001);
        //非循环贷款-当前逾期金额
        sb.append(Nsummary.getString("remainingOverdueAmount")).append(CRT001);
        //非循环贷款-当前最严重逾期状态
        sb.append(Nsummary.getString("remainingMaxOverdueStatus")).append(CRT001);
        //非循环贷款-累计贷款笔数
        sb.append(Nsummary.getString("loanCount")).append(CRT001);
        //非循环贷款-历史最严重逾期状态
        sb.append(Nsummary.getString("maxOverdueStatus")).append(CRT001);
        //非循环贷款-未结清贷款笔数
        sb.append(Nsummary.getString("openLoanCount")).append(CRT001);
        //非循环贷款-累计逾期次数
        sb.append(Nsummary.getString("overdueCount")).append(CRT001);
        //非循环贷款-最近一次发生代偿日期
        sb.append(Nsummary.getString("lastCompensationDate")).append(CRT001);
        //非循环贷款-近30天新增贷款机构数
        sb.append(ND30.getString("loanTenantCount")).append(CRT001);
        //非循环贷款-近30天平均贷款金额
        sb.append(ND30.getString("averageLoanAmount")).append(CRT001);
        //非循环贷款-近30天发生逾期贷款笔数
        sb.append(ND30.getString("overdueLoanCount")).append(CRT001);
        //非循环贷款-近30天新增贷款笔数
        sb.append(ND30.getString("loanCount")).append(CRT001);
        //非循环贷款-近30天最大贷款金额
        sb.append(ND30.getString("maxLoanAmount")).append(CRT001);
        //非循环贷款-近30天申请机构数量
        sb.append(ND30.getString("applyTenantCount")).append(CRT001);
        //非循环贷款-近30天新增贷款金额
        sb.append(ND30.getString("loanAmount")).append(CRT001);
        //非循环贷款-近30天发生代偿笔数
        sb.append(ND30.getString("compensationCount")).append(CRT001);
        //非循环贷款-近30天发生代偿次数
        sb.append(ND30.getString("compensationTimes")).append(CRT001);
        //非循环贷款-近30天发生代偿总金额
        sb.append(ND30.getString("compensationAmount")).append(CRT001);
        //非循环贷款-近90天新增贷款机构数
        sb.append(ND90.getString("loanTenantCount")).append(CRT001);
        //非循环贷款-近90天平均贷款金额
        sb.append(ND90.getString("averageLoanAmount")).append(CRT001);
        //非循环贷款-近90天发生逾期贷款笔数
        sb.append(ND90.getString("overdueLoanCount")).append(CRT001);
        //非循环贷款-近90天新增贷款笔数
        sb.append(ND90.getString("loanCount")).append(CRT001);
        //非循环贷款-近90天最大贷款金额
        sb.append(ND90.getString("maxLoanAmount")).append(CRT001);
        //非循环贷款-近90天申请机构数量
        sb.append(ND90.getString("applyTenantCount")).append(CRT001);
        //非循环贷款-近90天新增贷款金额
        sb.append(ND90.getString("loanAmount")).append(CRT001);
        //非循环贷款-近90天发生代偿笔数
        sb.append(ND90.getString("compensationCount")).append(CRT001);
        //非循环贷款-近90天发生代偿次数
        sb.append(ND90.getString("compensationTimes")).append(CRT001);
        //非循环贷款-近90天发生代偿总金额
        sb.append(ND90.getString("compensationAmount")).append(CRT001);
        //非循环贷款-近180天新增贷款机构数
        sb.append(ND180.getString("loanTenantCount")).append(CRT001);
        //非循环贷款-近180天平均贷款金额
        sb.append(ND180.getString("averageLoanAmount")).append(CRT001);
        //非循环贷款-近180天发生逾期贷款笔数
        sb.append(ND180.getString("overdueLoanCount")).append(CRT001);
        //非循环贷款-近180天新增贷款笔数
        sb.append(ND180.getString("loanCount")).append(CRT001);
        //非循环贷款-近180天最大贷款金额
        sb.append(ND180.getString("maxLoanAmount")).append(CRT001);
        //非循环贷款-近180天申请机构数量
        sb.append(ND180.getString("applyTenantCount")).append(CRT001);
        //非循环贷款-近180天新增贷款金额
        sb.append(ND180.getString("loanAmount")).append(CRT001);
        //非循环贷款-近180天发生代偿笔数
        sb.append(ND180.getString("compensationCount")).append(CRT001);
        //非循环贷款-近180天发生代偿次数
        sb.append(ND180.getString("compensationTimes")).append(CRT001);
        //非循环贷款-近180天发生代偿总金额
        sb.append(ND180.getString("compensationAmount")).append(CRT001);
        //非循环贷款-近360天新增贷款机构数
        sb.append(ND360.getString("loanTenantCount")).append(CRT001);
        //非循环贷款-近360天平均贷款金额
        sb.append(ND360.getString("averageLoanAmount")).append(CRT001);
        //非循环贷款-近360天发生逾期贷款笔数
        sb.append(ND360.getString("overdueLoanCount")).append(CRT001);
        //非循环贷款-近360天新增贷款笔数
        sb.append(ND360.getString("loanCount")).append(CRT001);
        //非循环贷款-近360天最大贷款金额
        sb.append(ND360.getString("maxLoanAmount")).append(CRT001);
        //非循环贷款-近360天申请机构数量
        sb.append(ND360.getString("applyTenantCount")).append(CRT001);
        //非循环贷款-近360天新增贷款金额
        sb.append(ND360.getString("loanAmount")).append(CRT001);
        //非循环贷款-近360天发生代偿笔数
        sb.append(ND360.getString("compensationCount")).append(CRT001);
        //非循环贷款-近360天发生代偿次数
        sb.append(ND360.getString("compensationTimes")).append(CRT001);
        //非循环贷款-近360天发生代偿总金额
        sb.append(ND360.getString("compensationAmount")).append(CRT001);
        //近12个月查询记录
        sb.append(info.getString("queryHistory")).append(CRT001);
        //报告编号
        sb.append(ReportHeader.getString("reportId")).append(CRT001);
        //查询结果
        sb.append(ReportHeader.getString("queryResult")).append(CRT001);
        //报告时间
        sb.append(ReportHeader.getString("reportTime")).append(CRT001);
        //姓名
        sb.append(PersonalProfile.getString("name")).append(CRT001);
        //手机号数量
        sb.append(PersonalProfile.getString("mobileCount")).append(CRT001);
        //身份证号码
        sb.append(PersonalProfile.getString("pid")).append(CRT001);
        //居住信息
        sb.append(info.getString("homeInfo")).append(CRT001);
        //职业信息
        sb.append(info.getString("workInfo")).append(CRT001);
        StringBuilder txSb = getBaseInfo(obj);
        key.set(getKey());
        val.set((sb.toString() + txSb.toString()).replace("\n", " ").replace("\r", " "));
        context.write(key, val);
        return true;
    }

    @Override
    public String getTableName() {
        return "kafka_babel_baihang_lend_inc";
    }

    @Override
    public String getType() {
        return RecordType.BAIHANG_C1_RECORD.toString();
    }

    @Override
    public String getComment() {
        return "百行征信";
    }
}
